using System;
using System.IO;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using Microsoft.BizTalk.TransportProxy.Interop;
using Microsoft.BizTalk.Message.Interop;
using Microsoft.XLANGs.BaseTypes;

namespace Microsoft.Samples.BizTalk.Adapter.Tcp.Receive
{
    /// <summary>
    ///     This class represents a worker thread that is used to dequeue work for a specific receive location
    ///     and submit in single message batches to the messaging engine in order.
    ///     This worker thread is not handled by the dispatcher, but directly by the receive location.
    /// </summary>
    /// <author>
    ///     Paul Brown
    /// </author>
    public class OrderedBatchWorkerThread : WorkerThread
    {
        #region Private Instance Members

        private ReceiveAdapter                          _receiveAdapter;    // The receive adapter
        private ReceiveLocation                         _receiveLocation;   // The ordered receive location running this worker thread
        private AutoResetEvent                          _orderedEvent;      // The wait object for ordering batches

        #endregion

        #region Constructor

        /// <summary>
        ///     This constructor initializes the instance.
        /// </summary>
        public OrderedBatchWorkerThread(IAdapter adapter)
            : base(adapter)
        {
            // Get thread name prefix
            this.ThreadNamePrefix = TraceHelper.ResourceManager.GetString(ResourceConstants.ReceiveOrderedBatchWorkerThreadName);

            // We know the adapter will be a receive adapter, store a reference to it
            this._receiveAdapter = (ReceiveAdapter)adapter;

            // Initialize
            this._receiveLocation = null;
            this._orderedEvent = new AutoResetEvent(false);
        }

        #endregion

        #region Public Instance Properties

        /// <summary>
        ///     This property returns a reference to the receive adapter.
        /// </summary>
        public ReceiveAdapter ReceiveAdapter
        {
            get { return (this._receiveAdapter); }
        }

        /// <summary>
        ///     This property gets or sets the receive location associated with this worker thread.
        /// </summary>
        public ReceiveLocation ReceiveLocation
        {
            get { return (this._receiveLocation); }
            set { this._receiveLocation = value; }
        }

        #endregion

        #region Private Instance Methods

        /// <summary>
        ///     This method creates a BizTalk message for submission into the batch.
        /// </summary>
        private IBaseMessage CreateMessage(Stream stream, IBaseMessageFactory messageFactory, OrderedBatchWorkItem workItem)
        {
            return (ReceiveLocation.CreateMessage(stream, messageFactory, workItem.ReceiveLocation));
        }

        /// <summary>
        ///     This method gets the data from the socket and closes the socket if it is a non-duplex receive location.
        ///     It has the potential for returning multiple streams depending on how many messages are stacked on the socket.
        /// </summary>
        private Stream[] GetData(OrderedBatchWorkItem workItem, out bool isLastStreamNotComplete)
        {
            // Receive the data
            bool lastStreamNotComplete;
            Stream[] streams = SocketIO.ReceiveData(this.Adapter,
                                                    workItem.ClientSocket,
                                                    workItem.ReceiveLocation.FrameStart,
                                                    workItem.ReceiveLocation.FrameEnd,
                                                    workItem.ReceiveLocation.DataEncoding,
                                                    this.ReceiveLocation.ReceiveDataBufferSize,
                                                    this.ReceiveLocation.AcceptMultipleMessages,
                                                    out lastStreamNotComplete);

            // Close the socket if it is a non-duplex one-way socket
            if ((!workItem.ReceiveLocation.IsDuplex) && (!workItem.ReceiveLocation.IsTwoWay))
            {
                if (workItem.ClientSocket != null)
                {
                    if (workItem.ClientSocket.Connected)
                    {
                        workItem.ClientSocket.Shutdown(SocketShutdown.Receive);
                        workItem.ClientSocket.Close(this.ReceiveLocation.MainCloseTimeout);
                    }
                }
            }

            // Return array
            isLastStreamNotComplete = lastStreamNotComplete;
            return (streams);
        }

        #endregion

        #region Abstract WorkerThread Members

        #region Public Instance Properties

        /// <summary>
        ///     This property returns the type of a OrderedBatchWorkItem which is compatible with this worker thread.
        /// </summary>
        public override Type WorkItemType
        {
            get { return (typeof(OrderedBatchWorkItem)); }
        }

        #endregion

        #region Public Instance Methods

        /// <summary>
        ///     This method overrides the base implementation and waits for the work item to complete before returning.
        /// </summary>
        public override void DoWork(IWorkItem workItem)
        {
            // Call base implementation which queues the work and releases the worker thread to do work
            base.DoWork(workItem);

            // Wait for the work to complete
            this._orderedEvent.WaitOne();
        }

        #endregion

        #region Protected Instance Methods

        /// <summary>
        ///     This method is the delegate that runs the main worker thread for ordered batches.  Work items are handled
        ///     individually in this worker thread so it doesn't control the batch through the normal batch size, data size
        ///     and batch timeout parameters.
        /// </summary>
        protected override void Thread()
        {
            // Variables
            bool submitted = false;
            bool                    threadInterrupted = false;
            int                     messageCount;
            OrderedBatchWorkItem    workItem = null;

            try
            {
                // Loop until we are interrupted
                while (true)
                {
                    // If we have work, start batching process
                    if (this.CurrentWorkLoad > 0)
                    {
                        if (this.Adapter.TraceSwitch.TraceVerbose)
                            TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveGetBatch, this.Name);

                        // Create a new batch for BizTalk
                        Batch batch = new Batch(this.Adapter);
                        batch.RetryCount = this.ReceiveAdapter.BatchRetryCount;
                        batch.RetryInterval = this.ReceiveAdapter.BatchRetryInterval;
                        batch.OrderedBatch = true;
                        messageCount = 0;

                        // Loop while we have work items
                        bool suspendAll = false;
                        while (this.CurrentWorkLoad > 0)
                        {
                            if (submitted == true)
                            {
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveInitialiseNewBatch, this.Name);
                                batch = new Batch(this.Adapter);
                                batch.RetryCount = this.ReceiveAdapter.BatchRetryCount;
                                batch.RetryInterval = this.ReceiveAdapter.BatchRetryInterval;
                                batch.OrderedBatch = true;
                                messageCount = 0;
                                submitted = false;
                            }
                            
                            // Dequeue item
                            workItem = (OrderedBatchWorkItem)this.DequeueWork();
                            this._receiveLocation = workItem.ReceiveLocation;

                            if (this.Adapter.TraceSwitch.TraceVerbose)
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveDequeuedWork, this.ReceiveLocation.Uri, this.Name);

                            // Set its status
                            workItem.Status = WorkItemStatusTypes.InProgress;

                            // Get data
                            bool isLastMessageNotComplete;
                            workItem.DataStreams = this.GetData(workItem, out isLastMessageNotComplete);

                            bool lastMessage = false;
                            for (int i = 0; i < workItem.DataStreams.Length; i++)
                            {
                                // Get stream
                                Stream dataStream = workItem.DataStreams[i];

                                // Is it the last message in the array?
                                if (i == (workItem.DataStreams.Length - 1))
                                {
                                    lastMessage = true;
                                }

                                if (dataStream != null)
                                {
                                    // Create a BizTalk message for the work item
                                    IBaseMessage message = this.CreateMessage(dataStream, this.Adapter.TransportProxy.GetMessageFactory(), workItem);

                                    bool addedSuccessfully = false;
                                    while (!addedSuccessfully)
                                    {
                                        try
                                        {
                                            // Add to the batch
                                            if (this.ReceiveLocation.IsTwoWay)
                                            {
                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmitRequestMessageToBatch, this.ReceiveLocation.Uri);

                                                // Set up response timeout value
                                                DateTime expirationTime = DateTime.Now.AddMinutes(this.ReceiveLocation.ResponseMessageTimeout);
                                                string correlationToken = Guid.NewGuid().ToString();

                                                // Submit to batch
                                                if ((lastMessage && isLastMessageNotComplete) || suspendAll)
                                                {
                                                    // Suspend message
                                                    BatchMessage batchMessage = new BatchMessage(message, new Exception(TraceHelper.FormatEntry(ResourceConstants.SocketSuspendedPartialMessage, dataStream.Length)));
                                                    batch.MoveToSuspendQ(batchMessage);

                                                    // Only set flag to suspend subsequent messages if the flag is set on the receive location
                                                    if (this.ReceiveLocation.SuspendSubsequentMessages)
                                                    {
                                                        suspendAll = true;
                                                    }
                                                }
                                                else
                                                {
                                                    // Add a transmitter to wait for the response
                                                    Transmitter transmitter = new Transmitter(this.Adapter, workItem.ReceiveLocation, workItem.ClientSocket);
                                                    Transmitter.AddHandler(transmitter);

                                                    // Submit message
                                                    batch.SubmitRequestMessage(new BatchMessage(message, correlationToken, expirationTime, transmitter));
                                                    
                                                    // Was the message suspended through failure to carry out operation?
                                                    if (batch.MessagesSuspended && this.ReceiveLocation.SuspendSubsequentMessages)
                                                    {
                                                        suspendAll = true;
                                                    }
                                                    else
                                                    {
                                                        // Wait for response
                                                        transmitter.WaitForResponse();
                                                    }
                                                }
                                            }
                                            else
                                            {
                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmitMessageToBatch, workItem.ReceiveLocation.Uri);

                                                // Submit to batch
                                                if ((lastMessage && isLastMessageNotComplete) || suspendAll)
                                                {
                                                    // Suspend message
                                                    BatchMessage batchMessage = new BatchMessage(message, new Exception(TraceHelper.FormatEntry(ResourceConstants.SocketSuspendedPartialMessage, dataStream.Length)));
                                                    batch.MoveToSuspendQ(batchMessage);

                                                    // Only set flag to suspend subsequent messages if the flag is set on the receive location
                                                    if (this.ReceiveLocation.SuspendSubsequentMessages)
                                                    {
                                                        suspendAll = true;
                                                    }
                                                }
                                                else
                                                {
                                                    // Submit message
                                                    batch.SubmitMessage(new BatchMessage(message));
                                                    
                                                    // Was the message suspended through failure to carry out operation?
                                                    if (batch.MessagesSuspended && this.ReceiveLocation.SuspendSubsequentMessages)
                                                    {
                                                        suspendAll = true;
                                                    }
                                                }
                                            }

                                            addedSuccessfully = true;
                                        }
                                        catch (ThreadInterruptedException)
                                        {
                                            // Asked to shut down, set flag and wait until this batch is done
                                            threadInterrupted = true;
                                        }
                                        catch (Exception e)
                                        {
                                            TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveFailedAddingToBatch, this.ReceiveLocation.Uri, batch.BatchCookie, e.Message);
                                            addedSuccessfully = false;
                                        }
                                    }

                                    // Increment message count
                                    messageCount++;
                                }
                            }

                            // Anything in batch?
                            if (messageCount > 0)
                            {
                                // Submit batch
                                //bool submitted = false;
                                for (int i = 0; i < this.ReceiveAdapter.BatchRetryCount; i++)
                                {
                                    try
                                    {
                                        if (this.Adapter.TraceSwitch.TraceVerbose)
                                            TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmittingBatch, messageCount);

                                        // Add to batch array
                                        if (Batch.GetBatch(batch.BatchCookie) == null)
                                        {
                                            Batch.AddBatch(batch);
                                        }

                                        // Submit batch (will block until IBTBatchCallback.BatchComplete called and completed)
                                        batch.SubmitBatch();

                                        // Did it ultimately fail?
                                        if (batch.BatchFailed)
                                        {
                                            // Shut down receive location to preserve order
                                            if (this.ReceiveLocation.SuspendSubsequentMessages)
                                            {
                                                Exception e = new Exception(TraceHelper.FormatEntry(ResourceConstants.BizTalkBatchFailed, batch.BatchCookie));
                                                this.Adapter.TransportProxy.ReceiverShuttingdown(this.ReceiveLocation.Uri, e);
                                            }
                                        }

                                        // Successful, break out of loop
                                        submitted = true;
                                        break;
                                    }
                                    catch (ThreadInterruptedException)
                                    {
                                        // Asked to shut down, set flag and wait until this batch is done
                                        threadInterrupted = true;
                                    }
                                    catch (Exception e)
                                    {
                                        // Trace
                                        TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveFailedSubmittingBatchRetrying, batch.BatchCookie, e.Message);

                                        // Wait interval
                                        System.Threading.Thread.Sleep(this.ReceiveAdapter.BatchRetryInterval * 1000);
                                    }
                                }

                                // If not successful and exhausted all retries
                                if (!submitted)
                                {
                                    string errorMessage = TraceHelper.FormatEntry(ResourceConstants.ReceiveFailedSubmittingBatch, batch.BatchCookie);
                                    TraceHelper.WriteRawEntry(TraceLevel.Error, errorMessage);
                                    Exception e = new Exception(errorMessage);

                                    // Save data
                                    workItem.Save();

                                    // Shut down receive location to preserve order if flag is set on receive location
                                    if (this.ReceiveLocation.SuspendSubsequentMessages)
                                    {
                                        this.Adapter.TransportProxy.ReceiverShuttingdown(this.ReceiveLocation.Uri, e);
                                    }
                                }
                            }
                            else
                            {
                                // No messages in batch!
                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveNoMessagesInBatch);
                            }

                            // Update work item
                            workItem.Status = WorkItemStatusTypes.Complete;
                            workItem = null;
                        }

                        // If the suspend all messages flag is set (that is, something suspended),
                        // shut down the receive location as there's no point receiving any more messages.
                        if (suspendAll)
                        {
                            // Set exception
                            string errorMessage = TraceHelper.FormatEntry(ResourceConstants.ReceiveStopReceivingMessagesToPreserveOrder, this.ReceiveLocation.Uri, batch.BatchCookie);
                            TraceHelper.WriteRawEntry(TraceLevel.Error, errorMessage);
                            Exception e = new Exception(errorMessage);

                            // Shut down receive location to preserve order
                            this.Adapter.TransportProxy.ReceiverShuttingdown(this.ReceiveLocation.Uri, e);
                        }
                    }

                    // Release DoWork thread (forces us to work 1 batch at a time to preserve order)
                    this._orderedEvent.Set();

                    // Was the thread interrupted?
                    if (threadInterrupted)
                    {
                        // Throw error again to end thread
                        throw (new ThreadInterruptedException());
                    }

                    if (this.Adapter.TraceSwitch.TraceVerbose)
                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadWaitingForWork, this.Name);

                    // Wait for signal that work has arrived
                    this.WaitForWork();
                }
            }
            catch (ThreadInterruptedException)
            {
                // Asked to shut down
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadInterrupted, this.Name);

                // If we have any in-flight work still, save it off
                if (workItem != null)
                {
                    workItem.Save();
                }

                while (this.CurrentWorkLoad > 0)
                {
                    workItem = (OrderedBatchWorkItem)this.DequeueWork();
                    workItem.Save();
                }

                // Release any DoWork thread
                this._orderedEvent.Set();
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.WorkerThreadGeneralError, this.Name, e.Message);

                // If we have an outstanding work item, push it back onto queue before restarting
                if (workItem != null)
                {
                    this.PushWorkOnQueue(workItem);
                }
            }
            finally
            {
                // Fire terminate event
                this.OnThreadTerminated(new EventArgs());
            }
        }

        #endregion

        #endregion
    }
}
